package com.key.w8.http1.connection.pool;

import com.key.w8.http1.connection.ConnectionBuilder;
import com.key.w8.http1.connection.connectionImpl.ClientHttp1PersistentConnection;
import com.key.w8.http1.connection.Connection;
import com.key.w8.http1.entity.HttpRequest;
import com.key.w8.http1.entity.HttpResponse;
import lombok.extern.slf4j.Slf4j;


import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @Author: k8
 * @CreateTime: 2025-03-02
 * @Version: 1.0
 *
 *
 * 定义应用层连接池,使用H1.1协议
 * 因为是针对单一服务器的多个连接，所以只需要最简单的轮询负载均衡算法
 *
 * 参照线程池但是又有所不同，线程池当无法再创建核心线程时先添加到工作队列，添加工作队列失败后再创建非核心
 * 为了减少所有连接使用一个工作队列会频繁的发生锁竞争，对每个队列使用独属的工作队列，再使用hash算法负载均衡每次返回哪个连接
 * 使用读写锁，每次使用hash算法需要获取连接数，有些非核心连接在关闭时需要拿到写锁然后关闭并修改连接数，为了避免返回一个即将关闭的连接
 * 当一个连接在尝试关闭时如果发现又被
 *
 *
 */
@Slf4j
public class ConnectionPool {
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;


    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }

    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
    private static int connectionCountOf(int c)  { return c & COUNT_MASK; }

    private static int ctlOf(int rs, int wc) { return rs | wc; }
    private boolean compareAndIncrementConnectionCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }

    /**
     * Attempts to CAS-decrement the workerCount field of ctl.
     */
    private boolean compareAndDecrementConnectionCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }
    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }

    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }


    /**
     * 最大连接数
     */
    private final int maxConnectionSize;


    private final int maxSendRequest;
    private ThreadFactory threadFactory;
    private ConnectionBuilder<ClientHttp1PersistentConnection,ConnectionPool.ConnectionWorker> builder;
    private ReentrantReadWriteLock.ReadLock mainRLock;
    private ReentrantReadWriteLock.WriteLock mainWLock;
    //todo 入口
    public ConnectionPool(int maxConnection,int maxSendRequest, ThreadFactory threadFactory, ClientConnectionBuilder builder) {
        assert maxSendRequest > 0;
        assert threadFactory != null;
        assert  builder != null;
        this.maxConnectionSize = maxConnection;
        this.maxSendRequest = maxSendRequest;
        this.connections = new ArrayList<>();
        this.threadFactory = threadFactory;
        this.builder = builder;
        ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock(true);
        mainRLock = reentrantReadWriteLock.readLock();
        mainWLock = reentrantReadWriteLock.writeLock();
        this.requestCounter = new AtomicInteger(0);
        this.index = new AtomicInteger(-1);
        this.connection2Worker = new HashMap<>(maxConnection);
    }

    private List<ConnectionWorker> connections;
    private HashMap<Connection,ConnectionWorker> connection2Worker;
    private final AtomicInteger requestCounter;
    private final AtomicInteger index;
    /*public ClientHttp1PersistentConnection getConnection(){
        int c = ctl.get();
        ClientHttp1PersistentConnection connection = null;
        if (connectionCountOf(c) < coreConnectionSize){
            connection = createCoreConnection(true);
        }
        if (connection == null){
            connections
        }
        return null;
    }*/
    /**
     * 在计算下一个发送请求的连接时，是无法获取写锁移除连接的，所以可以确保请求可以准确的被投递
     * shutDown只会修改标志位并调用所有connection的shutDown，如果connection支持继续执行，那么就继续执行
     * 并移除
     * @Author: k8
     * @date:
     * @param: httpRequest
     * @return: java.util.concurrent.Future<io.netty.handler.codec.http.HttpResponse>
     */
    public Future<HttpResponse> writeRequestNeedResponse(HttpRequest httpRequest){
        //避免shutDown时移除连接的同时又创建连接，使用读锁进行互斥
        int c = ctl.get();
        ClientHttp1PersistentConnection connection = null;
        if (!isRunning(c)){
            CompletableFuture<HttpResponse> result = new CompletableFuture<>();
            result.cancel(false);
            return result;
        }
        if (connectionCountOf(c) < maxConnectionSize){
            try {
                //在创建连接的时候会加写锁判断是否是存活状态，不是的话就不再添加
                connection = createCoreConnection();
            }catch (Throwable throwable){
                CompletableFuture<HttpResponse> result = new CompletableFuture<>();
                result.completeExceptionally(throwable);
                return result;
            }
        }
        if (connection != null){
            //能创建成功说明至少创建的时候线程池状态良好
            return connection.writeRequestNeedResponse(httpRequest);
        }else {
            try{
                mainRLock.lock();
                c = ctl.get();
                //在连接创建时，会先抢占资源，此时计数器的值可能并不是真实的连接数量，但是connections的插入和移除都是有写锁控制的
                //int counter = Math.min(connectionCountOf(c),connections.size());
                int counter = connections.size();
                int requests = requestCounter.get();
                if (requests+1 < maxSendRequest && isRunning(c)){
                    int i = index.get();
                    int newIndex = (i + 1) % counter;
                    while (!(index.compareAndSet(i,newIndex))){
                        i = index.get();
                        newIndex = (i + 1) % counter;
                    }
                    connection = connections.get(newIndex).connection;
                    return connection.writeRequestNeedResponse(httpRequest);
                }else {
                    CompletableFuture<HttpResponse> result = new CompletableFuture<>();
                    result.cancel(false);
                    return result;
                }
            }finally {
                mainRLock.unlock();
            }
        }
    }
    public void shutDown(){
        if (runStateAtLeast(ctl.get(),SHUTDOWN)){
            return;
        }
        try{
            mainWLock.lock();
            if (runStateAtLeast(ctl.get(),SHUTDOWN)){
                return;
            }
            int c = ctl.get();
            ctl.set(ctlOf(SHUTDOWN,connectionCountOf(c)));
        }finally {
            mainWLock.unlock();
        }
        List<ConnectionWorker> tempCs = new ArrayList<>(connections);
        tempCs.forEach(connectionWorker -> {
            try {
                connectionWorker.connection.shutDown();
            } catch (IOException e) {
            }
        });
    }
    //pool是先中断再关闭，这样的原因是关闭线程池时可能某个任务一直在等响应，而close必须保证任务被取消完，包括在执行任务，此时如果
    //不先中断让其提前结束，则close一直无法执行完，中断也一直没办法执行，虽然在得到响应后可以正常中断，但是不够快
    public void shutDownNow(){
        if (runStateAtLeast(ctl.get(),SHUTDOWN)){
            return;
        }
        try{
            mainWLock.lock();
            if (runStateAtLeast(ctl.get(),SHUTDOWN)){
                return;
            }
            int c = ctl.get();
            ctl.set(ctlOf(STOP,connectionCountOf(c)));
        }finally {
            mainWLock.unlock();
        }
        List<ConnectionWorker> tempCs = new ArrayList<>(connections);
        //放到外面来，避免死锁
        tempCs.forEach(connectionWorker -> {
            try {

                connectionWorker.thread.interrupt();
                connectionWorker.connection.close();
            } catch (IOException e) {
            }
        });
    }
    class ConnectionWorker{
        ClientHttp1PersistentConnection connection;
        Thread thread;
        Runnable connectListener;
        volatile boolean started;

        public void setConnectListener(Runnable connectListener) {
            this.connectListener = connectListener;
        }

        public ClientHttp1PersistentConnection getConnection() {
            return connection;
        }

        public void setConnection(ClientHttp1PersistentConnection connection) {
            this.connection = connection;
        }

        /**
         * 由tcp连接建立后的监听器回调调用
         *
         */
        public void start(){
            if (started) return;
            synchronized (this){
                if (started) return;
                started = true;
                thread.start();
            }
        }

        public void setThread(Thread thread) {
            this.thread = thread;
        }

        /**
         * 调用connect即开始准备连接，会调用在builder中setConnectListener添加的回调
         * 回调默认功能是开启连接，且setConnectListener执行是非阻塞的
         * 在监听器中会去建立连接并添加netty建立连接的监听器（该监听器负责调用run方法启动线程真正开启连接），当连接成功了就会调用监听器回调启动线程
         * 如此在创建connection时就不会阻塞等待tcp连接建立，而是在连接建立后在启动线程
         * 执行投递消息任务
         */
        void connect(){
            connectListener.run();
        }

        /**
         * 线程启动后调用，开启等待投递消息
         */
        void startConnection() {
            connection.runOut();
        }
    }
    private ClientHttp1PersistentConnection createCoreConnection() {
        int c1 = ctl.get();
        for (;;){
            if (runStateAtLeast(c1,SHUTDOWN)){
                return null;
            }
            if (connectionCountOf(c1) >= (maxConnectionSize & COUNT_MASK))
                return null;
            if (compareAndIncrementConnectionCount(c1)) break;
            c1 = ctl.get();
        }
        //c1++;
        boolean add = false;
        ConnectionWorker connectionWorker = new ConnectionWorker();
        Thread thread = threadFactory.newThread(() -> {
            connectionWorker.startConnection();
        });
        connectionWorker.setThread(thread);
        ClientHttp1PersistentConnection connection = builder.build(connectionWorker);
        try{
            mainWLock.lock();
            //创建连接必须是保证连接池状态是running，连接池状态修改需要加锁
            int c = ctl.get();
            if (runStateAtLeast(c,SHUTDOWN)) return null;
            if (thread.getState() != Thread.State.NEW) throw new IllegalStateException();
            //*************
            //添加回调，允许连接关闭时移除
            connection.addCloseListener(connection1 -> {
                try{
                    mainWLock.lock();
                    int c2 = ctl.get();
                    if (connection2Worker.containsKey(connection1)){
                        ConnectionWorker worker = connection2Worker.get(connection1);
                        connections.remove(worker);
                        connection2Worker.remove(connection1);
                    }
                    compareAndDecrementConnectionCount(c2);
                }finally {
                    mainWLock.unlock();
                }
            });
            //*******************
            connections.add(connectionWorker);
            connection2Worker.put(connection,connectionWorker);
            add = true;
        }finally {
            if (!add){
               ctl.addAndGet(-1);
            }
            mainWLock.unlock();
        }
        connectionWorker.connect();
        return connection;
    }

}
